package canal

import (
	"context"
	"sync/atomic"
	"time"

	logger "gitee.com/youkelike/zlog"
	"gitee.com/youkelike/zmigrate/migrator"
	"gitee.com/youkelike/zmigrate/migrator/events"
	"gitee.com/youkelike/zmigrate/migrator/validator"
	"gitee.com/youkelike/zmigrate/pkg/canalx"
	"gitee.com/youkelike/zmigrate/pkg/saramax"
	"github.com/IBM/sarama"
	"gorm.io/gorm"
)

// 消费由 canal 发出的 DB 变更消息，topic 为 zbook_binlog
type CanalConsumer[T migrator.Entity] struct {
	client    sarama.Client
	l         logger.Logger
	srcDBName string
	dstDBName string
	srcToDst  *validator.CanalIncrValidator[T]
	dstToSrc  *validator.CanalIncrValidator[T]
	dstFirst  *atomic.Bool
	runing    *atomic.Bool
	cg        sarama.ConsumerGroup
}

func NewCanalConsumer[T migrator.Entity](client sarama.Client, l logger.Logger, src, dst *gorm.DB, srcName, dstName string, p events.Producer) *CanalConsumer[T] {
	srcToDst := validator.NewCanalIncrValidator[T](src, dst, "SRC", l, p)
	dstToSrc := validator.NewCanalIncrValidator[T](src, dst, "DST", l, p)
	return &CanalConsumer[T]{
		client: client, l: l,
		srcDBName: srcName,
		dstDBName: dstName,
		dstFirst:  &atomic.Bool{},
		runing:    &atomic.Bool{},
		srcToDst:  srcToDst,
		dstToSrc:  dstToSrc}
}

func (r *CanalConsumer[T]) Start() (err error) {
	var t T

	if r.runing.Load() {
		r.l.Warnw("已经启动了", "CanalConsumer", t.TableName())
		return nil
	}

	topic := "zbook_binlog"
	groupName := "migrator_incr_" + t.TableName()
	// 每张要迁移的表对应一个独立的消费组
	r.cg, err = sarama.NewConsumerGroupFromClient(groupName, r.client)
	if err != nil {
		r.l.Errorw("启动增量校验失败",
			"err", err,
			"CanalConsumer", t.TableName())
		return err
	}
	r.runing.Store(true)
	go func() {
		err := r.cg.Consume(context.Background(),
			[]string{topic},
			saramax.NewCanalHandler[canalx.Message[T], T](r.l, r.Consume))
		if err != nil {
			r.l.Errorw("退出了消费循环异常",
				"err", err,
				"CanalConsumer", t.TableName())
		}
	}()
	r.l.Infow("启动 canal 增量校验消费者",
		"groupName", groupName,
		"topic", topic)
	return err
}

func (r *CanalConsumer[T]) Consume(msg *sarama.ConsumerMessage, val canalx.Message[T]) error {
	// 是不是源表为准
	dstFirst := r.dstFirst.Load()
	var v *validator.CanalIncrValidator[T]
	var t T
	// db:
	//  src:
	//    dsn: "root:root@tcp(localhost:13316)/zbook"
	//  dst:
	//    dsn: "root:root@tcp(localhost:13316)/zbook_intr"
	r.l.Debugw("用 canal 消息做增量校验", "消息内容", val)
	if val.Table != t.TableName() {
		return nil
	}
	if (val.Database != r.dstDBName) && (val.Database != r.srcDBName) {
		return nil
	}

	if val.Database == r.dstDBName && dstFirst {
		// 目标表为准
		// 校验，用 dst 的来校验
		v = r.dstToSrc
	} else if val.Database == r.srcDBName && !dstFirst {
		// 源表为准，并且消息恰好来自源表
		// 校验，用 src 来校验
		v = r.srcToDst
	} else {
		r.l.Warnw("校验消息和当前双写方向不一致",
			"dstFirst", dstFirst,
			"srcDBName", r.srcDBName,
			"dstDBName", r.dstDBName,
			"table", val.Table,
			"database", val.Database)
	}
	if v != nil {
		for _, data := range val.Data {
			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
			err := v.Validate(ctx, data.ID())
			cancel()
			if err != nil {
				r.l.Errorw("增量校验异常",
					"err", err,
					"CanalConsumer", t.TableName())
				return err
			}
		}
	}
	return nil
}

func (r *CanalConsumer[T]) DstFirst() {
	r.dstFirst.Store(true)
}
func (r *CanalConsumer[T]) Stop() error {
	var t T
	if !r.runing.Load() {
		r.l.Warnw("已经停止了", "CanalConsumer", t.TableName())
		return nil
	}

	err := r.cg.Close()
	if err != nil {
		r.l.Errorw("停止增量校验失败",
			"err", err,
			"CanalConsumer", t.TableName())
		return err
	}
	r.runing.Store(false)
	return nil
}
